package com.yxc.elasticsearchdemo;

import com.alibaba.fastjson.JSON;
import com.yxc.elasticsearchdemo.entiey.EsDemoEntity;
import com.yxc.elasticsearchdemo.entiey.EsDemoEntityBig;
import com.yxc.elasticsearchdemo.entiey.TestEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.apache.commons.lang3.StringUtils;
import org.apache.tomcat.util.bcel.Const;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author: 杨旭晨
 * @time: 2022/6/22 16:41
 **/
@Slf4j
@SpringBootTest
public class InsertDataToEs {

    @Autowired
    RestHighLevelClient client;
    @Autowired
    ThreadPoolTaskExecutor taskExecutor;

    /**
     * 插入数据
     *
     * @return void
     * @author 杨旭晨
     * @date 2022/6/23 17:55
     **/
    @Test
    void insert() throws IOException {
        for (Integer dataSize : Constant.dataSizeList) {
            BulkRequest bulkRequest = new BulkRequest();
            for (int i = 0; i < dataSize; i++) {
                EsDemoEntity esDemoEntity = new EsDemoEntity();
                esDemoEntity.setCName("更新测试" + dataSize);
                esDemoEntity.setCTime(LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
                esDemoEntity.setCValue("value");
                Map<String, Object> map = JSON.parseObject(JSON.toJSONString(esDemoEntity));
                bulkRequest.add(new IndexRequest(Constant.esIndex).source(map));
                if (bulkRequest.numberOfActions() >= 10000) {
                    // 10000条数据插入一次
                    client.bulk(bulkRequest, RequestOptions.DEFAULT);
                    bulkRequest = new BulkRequest();
                }
            }
            if (bulkRequest.numberOfActions() > 0) {
                // 插入剩下数据
                client.bulk(bulkRequest, RequestOptions.DEFAULT);
            }
            log.info("插入了{}条数据", dataSize);
        }
    }

    /**
     * 测试es是否可以使用大写字母开头作为属性值
     * 结果：不可以，使用RestHighLevelClient会自动转换成小写开头
     *
     * @author 杨旭晨
     * @date 2022/8/16 17:32
     * @return void
    **/
    @Test
    void insertTest() throws IOException {
        TestEntity testEntity = new TestEntity();
//        testEntity.setABC("ABC");
//        testEntity.setDEF("DEF");
//        testEntity.setHij("hij");
//        testEntity.setEND_TIME("END_TIME");
//        testEntity.setALARM_REASON("ALARM_REASON");
//        testEntity.setCONFIRM_STATUS("CONFIRM_STATUS");
//        testEntity.setGroupId("groupId");
//        testEntity.setReadState("readState");
        Map<String, Object> map = new HashMap<>();
        map.put("END_TIME", "end_time");
        map.put("ALARM_REASON", "alarm_reason");
        map.put("groupId", "groupId");
        map.put("readState", "readState");
        IndexRequest indexRequest = new IndexRequest("test_entity3");
        indexRequest.source(map);
//        indexRequest.source(JSON.parseObject(JSON.toJSONString(testEntity)));
//        System.out.println(JSON.parseObject(JSON.toJSONString(testEntity)));
        IndexResponse index = client.index(indexRequest, RequestOptions.DEFAULT);
    }

    @Test
    void test() throws IOException {
        BulkRequest bulkRequest = new BulkRequest();
        EsDemoEntity esDemoEntity = new EsDemoEntity();
        esDemoEntity.setCName("更新测试" + 200);
        esDemoEntity.setCTime(LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
        esDemoEntity.setCValue("value");
        Map<String, Object> map = JSON.parseObject(JSON.toJSONString(esDemoEntity));
        bulkRequest.add(new IndexRequest(Constant.esIndex).source(map));

        // 10000条数据插入一次
        client.bulk(bulkRequest, RequestOptions.DEFAULT);

    }

    @Test
    void insertBigData() throws IOException, InterruptedException {
//        String filePath = "/data/app/hunting/data/part-3-64";
        String filePath = "D:\\data\\sl\\part-3-64";
        File file = new File(filePath);
        LineIterator lineIterator = FileUtils.lineIterator(file);
        for (int i = 0; i < 20; i++) {
            taskExecutor.execute(() -> {
                BulkRequest bulkRequest = new BulkRequest();
                while (lineIterator.hasNext()) {
                    String logLine = lineIterator.next();
                    bulkRequest.add(new IndexRequest("big_data").source(JSON.parseObject(logLine), XContentType.JSON));
                    if (bulkRequest.numberOfActions() >= 1000) {
                        // 10000条数据插入一次
                        try {
                            BulkResponse bulk = client.bulk(bulkRequest, RequestOptions.DEFAULT);
                            System.out.println(bulk.status());
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        bulkRequest = new BulkRequest();
                    }
                }
                if (bulkRequest.numberOfActions() > 0) {
                    // 插入剩下数据
                    try {
                        client.bulk(bulkRequest, RequestOptions.DEFAULT);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        Thread.sleep(1000);
        while (true) {
            if (taskExecutor.getActiveCount() <= 0) {
                break;
            }
            Thread.sleep(1000);
        }
    }

    @Test
    void createIndex() throws IOException {
        CreateIndexRequest createIndexRequest = new CreateIndexRequest("big_data1");
        createIndexRequest.settings(
                Settings.builder().put("index.number_of_shards", 5).put("index.number_of_replicas", 1));
        Map<String, Object> properties = new HashMap<String, Object>();
        List<Map<String, Map<String, String>>> list = getProperties(EsDemoEntityBig.class);
        for (Map<String, Map<String, String>> map : list) {
            properties.putAll(map);
        }
        Map<String, Object> mapping = new HashMap<String, Object>();
        mapping.put("properties", properties);
        createIndexRequest.mapping(mapping);
        client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
    }

    public static List<Map<String, Map<String, String>>> getProperties(Class<?> clazz) {
        Field[] fields = clazz.getDeclaredFields();
        List<Map<String, Map<String, String>>> list = new ArrayList<>();
        for (Field field : fields) {
            Map<String, Map<String, String>> map = new HashMap<String, Map<String, String>>();
            Map<String, String> type = new HashMap<String, String>();
            String fieldName = field.getName();
            if (StringUtils.equals("log", fieldName) || StringUtils.equals("serialVersionUID", fieldName) || StringUtils.equals("id", fieldName)) {
                continue;
            }
            String fieldType = field.getAnnotation(org.springframework.data.elasticsearch.annotations.Field.class)
                    .type()
                    .name()
                    .toLowerCase();
            if (StringUtils.equals(fieldType, "auto")) {
                fieldType = "keyword";
            }
            type.put("type", fieldType);
            map.put(fieldName, type);
            list.add(map);
        }
        return list;
    }

}
